package com.jeffse.light.stream.core.function.base;

import com.jeffse.light.stream.core.data.context.KeyedContext;
import com.jeffse.light.stream.core.data.context.OperatorContext;
import com.jeffse.light.stream.core.data.context.ProcessContext;
import com.jeffse.light.stream.core.data.descriptor.stateDescriptor.ListStateDescriptor;
import com.jeffse.light.stream.core.data.descriptor.stateDescriptor.ValueStateDescriptor;
import com.jeffse.light.stream.core.data.state.ListState;
import com.jeffse.light.stream.core.data.state.ValueState;
import com.jeffse.light.stream.core.function.advance.SlideWindowFunction;
import com.jeffse.light.stream.core.pojo.window.SlideWindowInfo;
import com.jeffse.light.stream.core.tuple.Tuple2;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 窗口类型为固定次数窗口，滚动频率可按次数或者一段时间
 * @param <KEY>
 */
@Slf4j
public class CountSlideWindowKeyedProcessFunction<KEY, T, O>  extends KeyedProcessFunction<KEY, T, O> {
    private SlideWindowInfo slideWindowInfo = null;
    private SlideWindowFunction windowFunction = null;
    private WaterMarkFunction waterMarkFunction = null;
    private ValueState<Long> timestampState;
    private ListState<Tuple2<Long, T>> listState;
    private ListState<Tuple2<Long, T>> incrementListState;
    private ValueState<Boolean> isTrigger;
    private ValueState<Tuple2<Long, T>> valueState;

    public CountSlideWindowKeyedProcessFunction(SlideWindowInfo slideWindowInfo, SlideWindowFunction windowFunction, WaterMarkFunction waterMarkFunction) throws Exception {
        this.slideWindowInfo = slideWindowInfo;
        this.windowFunction = windowFunction;
        this.waterMarkFunction = waterMarkFunction;
        if(slideWindowInfo.getIsEventTime() && windowFunction == null){
            log.error("事件时间窗口必须设置 waterMark");
            throw new Exception("事件时间窗口必须设置 waterMark");
        }
    }

    @Override
    public void open(){
        super.open();
        timestampState = getRuntimeContext().getState(new ValueStateDescriptor("timestampState", this));
        isTrigger = getRuntimeContext().getState(new ValueStateDescriptor("isTrigger", this));
        valueState = getRuntimeContext().getState(new ValueStateDescriptor("valueState", this));
        listState = getRuntimeContext().getListState(new ListStateDescriptor("listState", this));
        incrementListState = getRuntimeContext().getListState(new ListStateDescriptor("incrementListState", this));
    }

    @Override
    public void processElement(ProcessContext context, T value) throws Exception {
        Tuple2<Long, T> item = new Tuple2<>(System.currentTimeMillis(), value);
        valueState.update(item);
        handleIncrementList(item);
        handleList(item);
    }

    @Override
    public void onTimer(long timestamp, KeyedContext ctx, ProcessContext sourceContext) throws Exception {
        super.onTimer(timestamp, ctx, sourceContext);
        compute();
        // 滚动窗口类型为时间类型，并且滚动窗口为系统时间， 在此设置下一次滚动时间
        if(!slideWindowInfo.getUnitY().equals(OperatorContext.SlideType.COUNT) && !slideWindowInfo.getIsEventTime()){
            Long time = System.currentTimeMillis() + slideWindowInfo.getY() * 1000;
            getTimerService().registerProcessingTimeTimer(time);
            timestampState.update(time);
        }
    }

    private void handleList(Tuple2<Long, T> item){
        List<Tuple2<Long, T>> list = new ArrayList<>();
        if(listState.getList() != null){
            list.addAll(listState.getList());
        }

        list.add(item);
        listState.update(list);
        if(list.size() >= slideWindowInfo.getX() && isTrigger.value() == null){
            isTrigger.update(true);
            compute();

            // 滚动窗口类型为时间类型，并且滚动窗口为系统时间， 在此设置第一次滚动时间
            if(!slideWindowInfo.getUnitY().equals(OperatorContext.SlideType.COUNT) && !slideWindowInfo.getIsEventTime()){
                Long time = System.currentTimeMillis() + slideWindowInfo.getY() * 1000;
                getTimerService().registerProcessingTimeTimer(time);
                timestampState.update(time);
            }

        }
    }

    private void handleIncrementList(Tuple2<Long, T> item){
        if(isTrigger.value() != null && isTrigger.value()){
            List<Tuple2<Long, T>> incrementList = new ArrayList<>();
            if(incrementListState.getList() != null){
                incrementList.addAll(incrementListState.getList());
            }

            incrementList.add(item);
            incrementListState.update(incrementList);

            // 次数滚动窗口，每隔 y 次数据滚动一次，只要增量数据大于等于 slideWindowInfo 设定的 y 的数量即可触发计算
            if(slideWindowInfo.getUnitY().equals(OperatorContext.SlideType.COUNT) && slideWindowInfo.getY() <= incrementList.size()){
                compute();
            }

            // 时间滚动窗口， 每隔 y 秒滚动一次(系统时间滚动窗口，跟事件无关，每次触发后，更新下次触发时间)
            if(!slideWindowInfo.getUnitY().equals(OperatorContext.SlideType.COUNT) && slideWindowInfo.getIsEventTime()){
                Long time = waterMarkFunction.setWaterMark(valueState.value().f1).getTimestamp();
                if(timestampState.value() == null || time - slideWindowInfo.getY() * 1000 >= timestampState.value()){
                    compute();
                    timestampState.update(time);
                }
            }
        }
    }

    public void compute() {
        List<Tuple2<Long, T>> list = new ArrayList<>();
        if(waterMarkFunction != null){
            list = listState.getList().stream().filter(item-> {
                return waterMarkFunction.setWaterMark(item.f1).getTimestamp() > waterMarkFunction.setWaterMark(valueState.value().f1).getTimestamp() - slideWindowInfo.getX() * 1000;
            }).collect(Collectors.toList());
        }else{
            list = listState.getList().stream().filter(item-> {
                return item.f0 > System.currentTimeMillis() - slideWindowInfo.getX() * 1000;
            }).collect(Collectors.toList());
        }

        List<T> computeList = list.stream().map(item->item.f1).collect(Collectors.toList());
        windowFunction.window(computeList, this);
        listState.update(list);
        incrementListState.clear();
    }
}